package com.cw.spider.queue;

import cn.wanghaomiao.seimi.annotation.Queue;
import cn.wanghaomiao.seimi.core.SeimiProcessor;
import cn.wanghaomiao.seimi.core.SeimiQueue;
import cn.wanghaomiao.seimi.struct.Request;
import com.google.common.base.Charsets;
import com.google.common.collect.Maps;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnel;
import com.google.common.hash.PrimitiveSink;

import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 描述:
 *
 * @author withqianqian@163.com
 * @create 2020-08-27 15:19
 */
@Queue
public class Myqueue implements SeimiQueue {
    private ExecutorService workersPool;
    private volatile BlockingQueue<Request> queue = new LinkedBlockingQueue<Request>();
    final public Map<String, SeimiProcessor> processorMap = Maps.newConcurrentMap();
    final public Map<String, Integer> container = Maps.newConcurrentMap();
    private final BloomFilter<String> dealIdBloomFilter = BloomFilter.create(new Funnel<String>() {

        private static final long serialVersionUID = 1L;

        @Override
        public void funnel(String arg0, PrimitiveSink arg1) {

            arg1.putString(arg0, Charsets.UTF_8);
        }

    }, 1024 * 1024 * 32);

    /**
     * 阻塞式出队一个请求
     *
     * @param crawlerName --
     * @return --
     */
    @Override
    public Request bPop(String crawlerName) {
        try {
            return queue.poll(10, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

    public Request bPop() {
        try {
            return queue.poll(10, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 入队一个请求
     *
     * @param req 请求
     * @return --
     */
    @Override
    public boolean push(Request req) {
        try {
            queue.put(req);
            return true;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 任务队列剩余长度
     *
     * @param crawlerName --
     * @return num
     */
    @Override
    public long len(String crawlerName) {
        return queue.size();
    }

    /**
     * 判断一个URL是否处理过了
     *
     * @param req --
     * @return --
     */
    @Override
    public boolean isProcessed(Request req) {
        String url = req.getUrl();
        if (!dealIdBloomFilter.mightContain(url)) {
            dealIdBloomFilter.put(url);

        } else {
            return true;
        }
        return false;
    }

    /**
     * 记录一个处理过的请求
     *
     * @param req --
     */
    @Override
    public void addProcessed(Request req) {
        if (container.get(req.getCrawlerName()) != null) {
            int incr = container.get(req.getCrawlerName());
            container.put(req.getCrawlerName(), ++incr);
        } else {
            container.put(req.getCrawlerName(), 0);
        }
    }

    /**
     * 目前总共的抓取数量
     *
     * @param crawlerName --
     * @return num
     */
    @Override
    public long totalCrawled(String crawlerName) {

        return container.get(crawlerName) != null ? container.get(crawlerName) : 0;
    }

    /**
     * 清除抓取记录
     *
     * @param crawlerName --
     */
    @Override
    public void clearRecord(String crawlerName) {
        container.put(crawlerName, 0);
    }

    public void run() {
        while (true) {
            Request request = bPop();
            if (request != null) {
                workersPool.execute(() -> {
                    processorMap.get(request.getCrawlerName()).run(request);
                });
            }


        }
    }

    public void setWorkersPool(ExecutorService workersPool) {
        this.workersPool = workersPool;
    }

}
